package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


//数据流：web/app -> 日志文件 -> flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序: Mock -> 文件 -> flume -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //  状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //2.从Kafka 读取topic_log主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("topic_log", "base_log_app"), WatermarkStrategy.noWatermarks(), "kafka-source");
        //3.过滤null值，并将数据转换为JSONOBJECT
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                if (s != null) {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                }
            }
        });

        //jsonObjDS.print("json-->");

        //4.按照Mid Keyby
        KeyedStream<JSONObject, String> keyedMidStream = jsonObjDS.keyBy(r -> r.getJSONObject("common").getString("mid"));
        //5.按照新老用户调整is_new字段值
        SingleOutputStreamOperator<JSONObject> jsonObjWithIsNewDS = keyedMidStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("mid-state", String.class);
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                //获取状态
                String lastDt = valueState.value();
                Long ts = jsonObject.getLong("ts");
                String curDt = DateFormatUtil.toYmdHms(ts);
                String isNew = jsonObject.getJSONObject("common").getString("is_new");

                if ("1".equals(isNew)) {
                    if (lastDt == null || lastDt.equals(curDt)) {
                        valueState.update(curDt);

                    } else {
                        jsonObject.getJSONObject("common").put("is_new", "0");
                    }
                } else {
                    if (lastDt == null) {
                        valueState.update("1970-01-01");
                    }
                }
                return jsonObject;
            }
        });

//        jsonObjWithIsNewDS.print("jsonObjWithIsNewDS-->");
//
        //6.使用测输出流分流page,displays,actions,start,error,appVideo
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displaysTag = new OutputTag<String>("displays") {
        };
        OutputTag<String> actionsTag = new OutputTag<String>("actions") {
        };
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideo") {
        };

        SingleOutputStreamOperator<String> pageDS = jsonObjWithIsNewDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //尝试获取error数据
                String err = jsonObject.getString("err");
                if (err != null) {
                    context.output(errorTag, jsonObject.toJSONString());
                    jsonObject.remove("err");
                }


                //尝试获取start数据
                String start = jsonObject.getString("start");
                if (start != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                    if (appVideo != null) {
                        context.output(appVideoTag, jsonObject.toJSONString());
                        jsonObject.remove("appVideo");
                    } else {
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        Long ts = jsonObject.getLong("ts");

                        //获取displays数据
                        JSONArray displays = jsonObject.getJSONArray("displays");
                        if (displays != null && displays.size() > 0) {
                            for (int i = 0; i < displays.size(); i++) {
                                JSONObject display = displays.getJSONObject(i);
                                //补充page_id,ts字段
                                display.put("page_id", pageId);
                                display.put("ts", ts);
                                context.output(displaysTag, display.toJSONString());
                            }
                        }
                        //获取actions数据
                        JSONArray actions = jsonObject.getJSONArray("actions");
                        if (actions != null && actions.size() > 0) {
                            for (int i = 0; i < actions.size(); i++) {
                                JSONObject action = actions.getJSONObject(i);
                                //补充page_id字段
                                action.put("page_id", pageId);
                                context.output(actionsTag, action.toJSONString());
                                jsonObject.remove("displays");
                                jsonObject.remove("actions");
                                collector.collect(jsonObject.toJSONString());

                            }
                        }
                        //获取appVideo数据
                    }
                }

//                if (){
//
//                }else if (){
//
//                }else {
//
//                }
//
//
//                if () {
//                }
//
//                if (){
//                }else {
//                }
            }
        });
        //7.获取分流数据
        DataStream<String> errDS = pageDS.getSideOutput(errorTag);
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displaysDS = pageDS.getSideOutput(displaysTag);
        DataStream<String> actionsDS = pageDS.getSideOutput(actionsTag);
        DataStream<String> appVideoDS = pageDS.getSideOutput(appVideoTag);

        errDS.print("errDS-->");
        startDS.print("startDS-->");
        displaysDS.print("displaysDS-->");
        actionsDS.print("actionsDS-->");
        appVideoDS.print("appVideoDS-->");
        pageDS.print("pageDS-->");


        //8.写出到kafka
        String page_topic = "dwd_traffic_page_view";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String appVideo_topic = "dwd_traffic_video_play";


        pageDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        errDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displaysDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionsDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        appVideoDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(appVideo_topic));
        //9.执行
        env.execute("BaseLogApp");
    }
}
